package com.hub.window;

import com.hub.wc.MyFlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.OutputTag;

public class OtherFromSocket {

    public static void main(String[] args) throws Exception {


        // 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 数据集
        DataStream<String> inputDataStream = env.socketTextStream("119.23.58.27", 8082);

        //
        OutputTag<Tuple2<String, Integer>> outputTag = new OutputTag<Tuple2<String, Integer>>("after 30s") {};

//        DataStream<Tuple2<String, Integer>> output = inputDataStream
//                .flatMap(new MyFlatMapFunction())
//                .keyBy(0)
//                .sum(1);

        SingleOutputStreamOperator<Tuple2<String, Integer>> sumStream = inputDataStream.flatMap(new MyFlatMapFunction())
                                                                        .keyBy(0)
                                                                        .timeWindow(Time.seconds(15))
                                                                        .allowedLateness(Time.seconds(30))
                                                                        .sideOutputLateData(outputTag)
                                                                        .sum(1);

        sumStream.getSideOutput(outputTag).print("between 10~15");
        sumStream.print("inner 10s");
        env.execute();

    }

}
